package ai.people.core.standard.gather.userconfig;

import ai.people.core.common.util.ApplicationContextUtil;
import ai.people.core.standard.gather.entity.UserConfigKeywordData;
import ai.people.netmon.framework.domain.gather.config.GatherSyncExtended;
import ai.people.core.standard.gather.entity.UserConfigAccountData;
import ai.people.netmon.framework.domain.gather.response.GatherResponseBody;
import ai.people.netmon.framework.domain.userconfig.request.SyncDataRequest;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * 用户配置-账号同步抽取服务类
 *
 * @author zhenglin
 * @since 2022/3/9
 */
public interface UserConfigSyncService<T> {
    Logger LOGGER = LoggerFactory.getLogger(UserConfigSyncService.class);


    /**
     * 同步账户
     *
     * @param data        帐户数据
     * @param sourceType         源类型
     * @param topic              主题
     * @param kafkaTemplate      卡夫卡模板
     * @param gatherSyncExtended 收集同步扩展
     * @param updateSupplier     更新供应商
     */
    default void syncAccounts(UserConfigAccountData data, String sourceType, String topic, KafkaTemplate kafkaTemplate, GatherSyncExtended gatherSyncExtended, Supplier updateSupplier){
        syncAccounts(Collections.singletonList(data),sourceType,topic,kafkaTemplate,gatherSyncExtended,updateSupplier);
    }

    /**
     * 同步账户
     *
     * @param list               列表
     * @param sourceType         源类型
     * @param topic              主题
     * @param kafkaTemplate      卡夫卡模板
     * @param gatherSyncExtended 收集同步扩展
     * @param updateSupplier     更新供应商
     */
    default void syncAccounts(List<UserConfigAccountData> list, String sourceType, String topic, KafkaTemplate kafkaTemplate, GatherSyncExtended gatherSyncExtended, Supplier updateSupplier){
       sync(list,sourceType,topic,kafkaTemplate,gatherSyncExtended,updateSupplier);
    }


    /**
     * 同步关键字
     *
     * @param data               数据
     * @param sourceType         源类型
     * @param topic              主题
     * @param kafkaTemplate      卡夫卡模板
     * @param gatherSyncExtended 收集同步扩展
     * @param updateSupplier     更新供应商
     */
    default void syncKeywords(UserConfigKeywordData data, String sourceType, String topic, KafkaTemplate kafkaTemplate, GatherSyncExtended gatherSyncExtended, Supplier updateSupplier){
        syncKeywords(Collections.singletonList(data),sourceType,topic,kafkaTemplate,gatherSyncExtended,updateSupplier);
    }

    /**
     * 同步关键字
     *
     * @param data               数据
     * @param sourceType         源类型
     * @param topic              主题
     * @param kafkaTemplate      卡夫卡模板
     * @param gatherSyncExtended 收集同步扩展
     * @param updateSupplier     更新供应商
     */
    default void syncKeywords(List<UserConfigKeywordData> data, String sourceType, String topic, KafkaTemplate kafkaTemplate, GatherSyncExtended gatherSyncExtended, Supplier updateSupplier){
        sync(data,sourceType,topic,kafkaTemplate,gatherSyncExtended,updateSupplier);
    }

    /**
     * 同步
     *
     * @param list               列表
     * @param sourceType         源类型
     * @param topic              主题
     * @param kafkaTemplate      卡夫卡模板
     * @param gatherSyncExtended 收集同步扩展
     * @param updateSupplier     更新供应商
     */
    default <E> void sync(List<E> list, String sourceType, String topic, KafkaTemplate kafkaTemplate, GatherSyncExtended gatherSyncExtended, Supplier updateSupplier){
        ExecutorService pool = ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor");
        pool.execute(()->{
            List<List<E>> partition = Lists.partition(list, gatherSyncExtended.getLimit());
            for (List<E> part : partition) {
                SyncDataRequest<E> ori = new SyncDataRequest<>(part, sourceType);
                String msg = JSONObject.toJSONString(ori);
                kafkaTemplate.send(topic,msg);
                LOGGER.info("推送topic:{} 数据条数:{}",topic,part.size());
            }
            updateSupplier.get();
        });
    }

    /**
     * 同步账户
     *
     * 新方法参考
     * @see  #syncAccounts
     *
     * @param data               数据
     * @param type               类型
     * @param url                url
     * @param gatherSyncExtended 收集同步扩展
     */
    @Deprecated
    default void syncAccounts(List<T> data, String type, String url, GatherSyncExtended gatherSyncExtended) {
        ExecutorService pool = ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor");
        RestTemplate restTemplate = ApplicationContextUtil.getBean(RestTemplate.class);
        pool.execute(() -> {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            Map<String, String> pidMapping = gatherSyncExtended.getPidMapping();
            List<List<T>> partition = Lists.partition(data, gatherSyncExtended.getLimit());
            for (List<T> part : partition) {
                List<UserConfigAccountData> userConfigAccountData = part.stream().map(x -> JSONObject.parseObject(JSONObject.toJSONString(x), UserConfigAccountData.class)).collect(Collectors.toList());
                userConfigAccountData.forEach(x->x.setPartnerId(pidMapping.getOrDefault(x.getAccount(),"[]")));
                SyncDataRequest<UserConfigAccountData> body = new SyncDataRequest<>(userConfigAccountData, type);
                HttpEntity<SyncDataRequest<UserConfigAccountData>> request = new HttpEntity<>(body, headers);
                LOGGER.info("请求url:{} 数据条数:{} 等待返回...", url, part.size());
                for (int i = 0; i < gatherSyncExtended.getNumberOfRetries(); i++) {
                    GatherResponseBody result = null;
                    try {
                        result = restTemplate.postForObject(url, request, GatherResponseBody.class);
                    } catch (Exception e) {
                        LOGGER.info("重试...");
                        LOGGER.error(e.getMessage(), e);
                        continue;
                    }
                    LOGGER.info("请求url:{}数据同步请求返回:{}", url, result);
                    //更新数据为已采集状态
                    Integer status = Optional.ofNullable(result).map(GatherResponseBody::getStatus).orElse(null);
                    if (Objects.equals(status, CommonEnum.SUCCESS.getCode())) {
                        this.updateCollectionStatus(part);
                        break;
                    }
                }
            }
        });
    }

    /**
     * 更新采集状态
     *
     * @param account 账户
     */
    @Deprecated
    void updateCollectionStatus(List<T> account);
}
